package com.heima.article.kafka;

import com.alibaba.fastjson.JSON;
import com.heima.article.controller.v1.AutomaticController;
import com.heima.model.article.dto.CollectionBehaviorDto;
import com.heima.article.service.ApCollectionService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;

import java.util.Optional;

//@Component
public class KfkListener {

    @Autowired
    AutomaticController automaticController;

    @KafkaListener(topics = {"autoScanByMediaNewsId"})
    public void listener(ConsumerRecord<?, ?> record) {
       System.out.println(record.key() + "===========" + record.value());
        /*Object value = record.value();
        automaticController.autoScanByMediaNewsId(Integer.valueOf((String)value));*/

        Optional<? extends ConsumerRecord<?, ?>> optional = Optional.ofNullable(record); //先判断是否为空
        if(optional.isPresent()){
            Object value = record.value();
            automaticController.autoScanByMediaNewsId(Integer.valueOf((String) value));
        }
    }

    @Autowired
    ApCollectionService apCollectionService;
    @KafkaListener(topics = {"collection"})
    public void collectionlistener(ConsumerRecord<?, ?> record) {
        System.out.println(record.key() + "===========" + record.value());
        /*Object value = record.value();
        automaticController.autoScanByMediaNewsId(Integer.valueOf((String)value));*/

        Optional<? extends ConsumerRecord<?, ?>> optional = Optional.ofNullable(record); //先判断是否为空
        if(optional.isPresent()){
            Object value = record.value();
            apCollectionService.collection(JSON.parseObject(value.toString(), CollectionBehaviorDto.class));
        }
    }
}